<?php
namespace Tbf\Bundle\QueueBundle;
use Tbf\Bundle\QueueBundle\Entity\QueueData;
use Doctrine\DBAL\LockMode;
use Doctrine\ORM\EntityManager;
class Queue
{
    protected $name;
    /**
     * @var EntityManager
     */
    protected $om;
    public function __construct($name,EntityManager $om)
    {
        $this->name = $name;
        $this->om = $om;
    }
    /**
     * 向这个消息队列写入一条数据
     * @param interface{} $data
     */
    public function write($data)
    {
        $queue = new QueueData($this->name,$data);
        $this->om->persist($queue);
        $this->om->flush();
    }

    /**
     * 从这个消息队列读出一条数据,然后这条数据就会被删除
     * 如果没有更多数据,err会返回EOF
     * @return interface{} $data
     * @return string      $error
     */
    public function read()
    {
        $this->om->beginTransaction();
        $queue_entity = $this->om->createQuery('select q from TbfQueueBundle:QueueData q where q.channel=:channel order by q.id DESC')
            ->setMaxResults(1)
            ->setLockMode(LockMode::PESSIMISTIC_WRITE)
            ->execute(array('channel'=>$this->name));
        if (empty($queue_entity)) {
            $this->om->rollback();

            return array(null,'EOF');
        }
        $data = $queue_entity[0]->getData();
        $this->om->remove($queue_entity[0]);
        $this->om->flush();
        $this->om->commit();

        return array($data,null);
    }
}
